﻿
using System;
using System.Threading;
using System.IO;

namespace MX.Core.Caching.Redis.Sider.Executors
{
	internal class MonitorExecutor : ExecutorBase
	{
		private Observable<string> _observable;

    
		public override Mode Mode { get { return Mode.Monitor; } }

		public override T Execute<T> (Invocation<T>  invocation, Stream stream)
		{
			if (invocation.Command == "QUIT") {
				// QUIT issued by the observable automatically
				_observable.Dispose ();
				Client.Dispose ();
				return default(T);
			}

			throw new InvalidOperationException (
				"Only QUIT for terminating MONITOR is allowed.");
		}

		public IObservable<string> BuildObservable (Stream _stream)
		{
			return _observable = new MonitorObservable (_stream, this);
		}


		private class MonitorObservable : Observable<string>
		{
			private MonitorExecutor _parent;
			Stream _stream;

			public MonitorObservable (Stream stream, MonitorExecutor parent)
			{
				_stream = stream;
				_parent = parent;
				ThreadPool.QueueUserWorkItem (monitorThread);
			}

			private void monitorThread (object _)
			{
				try {
					while (!IsDisposed)
						Next (_parent.Reader.ReadStatus (_stream));

					Complete ();
				} catch (Exception e) {
					Error (e);
				}
			}


			public override void Dispose ()
			{
				_parent.ExecuteImmediate (Invocation<Object>.New ("QUIT", (w, s) => {
				}, (r, s) => null), _stream);
				//to  close true
			}
		}
	}
}
